package com.junbao.message.infrastructure.core.dispatcher;

import java.util.concurrent.ThreadPoolExecutor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.commom.context.ClassPathXmlApplicationContext;
import com.junbao.message.application.service.MessageService;
import com.junbao.message.domain.model.NotifyMessage;
import com.junbao.message.infrastructure.core.AbstractDisruptorCore;
import com.junbao.message.infrastructure.core.DataConsumer;
import com.junbao.message.infrastructure.core.DataProducer;
import com.junbao.message.infrastructure.core.RequestDto;
import com.junbao.message.infrastructure.core.handler.MessageDeadHandler;
import com.junbao.message.infrastructure.core.producer.MessageDeadProducer;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.WorkProcessor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 死亡消息disruptor
 * @author yilin
 *
 * @param <T>
 */
@SuppressWarnings("all")
public class MessageDeadDisruptor extends AbstractDisruptorCore{

	private static final Logger LOGGER = LoggerFactory.getLogger(MessageDeadDisruptor.class);
	
	private static class DisruptorCoreInstance{
		private final static MessageDeadDisruptor INSTANCE = new MessageDeadDisruptor();
	}
	
	private MessageDeadDisruptor(){};
	
	public static MessageDeadDisruptor instance(){
		return DisruptorCoreInstance.INSTANCE;
	}
	
	static{
		try {
			MessageDeadDisruptor disruptorCore = MessageDeadDisruptor.instance();
			//获取消费者
			DataConsumer[] messageConsumers = getmessageConsumers(disruptorCore.setCounsumerClassType(),AbstractDisruptorCore.DEFAULT_CONSUMER_LENGTH);
			//启动disruptor
			disruptorCore.start(ProducerType.MULTI,
					RINGBUFFER_SIZE,
					new BlockingWaitStrategy(),
					messageConsumers,
					RequestDto.class);
		} catch (Exception e) {
			LOGGER.error("启动MessageDeadDisruptor失败！",e);
		}
	}

	@Override
	public DataProducer createDataProducer(String producerName) {
		return new MessageDeadProducer(producerName, this.ringBuffer);
	}
	
	@Override
	public String setWorkThreadName() {
		return "MessageDead-thread";
	}

	@Override
	public Class setCounsumerClassType() {
		return MessageDeadHandler.class;
	}
	
	@Override
	public void rejectedExecutionHandler(Runnable runnable, ThreadPoolExecutor executor) {
		executor.execute(runnable);
	}

	@Override
	public void disruptorExceptionHandler(Throwable throwable, long sequence, Object event) {
		RequestDto<NotifyMessage> requestDto =(RequestDto)event;
		LOGGER.error("消息修改死亡状态失败，消息编号："+requestDto.getBusinessObject().getMessageNo(),throwable);
		MessageService messageService = (MessageService) ClassPathXmlApplicationContext.getInstance().getBean("messageService");
		messageService.updateMessageToDead(requestDto.getBusinessObject());
	}
}
